package com.magupe.im.web.db.service.impl;

import com.magupe.im.web.db.service.DataService;
import com.magupe.im.web.protocol.request.GroupMessageRequest;
import com.magupe.im.web.protocol.request.MessageRequest;
import com.magupe.im.web.protocol.response.StatusCode;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import org.apache.commons.collections.CollectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;

public abstract class DataServiceImpl implements DataService {

    protected Logger logger = LoggerFactory.getLogger(DataServiceImpl.class);

    @Override
    public abstract void saveMessage(MessageRequest request);

    @Override
    public abstract void saveGroupMessage(GroupMessageRequest request);

    @Override
    public final void sendOfflineMessage(String appId, String receiverId, String groupId, Channel channel) {
        List<MessageRequest> messages = getMessages(appId, receiverId);
        List<GroupMessageRequest> groupMessages = getGroupMessages(appId, groupId);

        sendMessage(messages, channel);
        sendGroupMessage(groupMessages, channel);
    }

    @Override
    public void cacheMessage(ChannelFuture future, MessageRequest request) {
        future.addListener(new ChannelFutureListener(){
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                if (!future.isSuccess() || future.isCancelled()) {
                    saveMessage(request);
                }
            }
        });
    }

    public final void sendMessage(List<MessageRequest> messages, Channel channel){
        if(!CollectionUtils.isEmpty(messages)){
            messages.forEach((message) -> { message.setStatusCode(StatusCode.MESSAGE_SEND_SUCCESS); });
            messages.forEach((message) -> {
                ChannelFuture future = channel.writeAndFlush(new TextWebSocketFrame(message.toString()));
                future.addListener(new ChannelFutureListener(){
                    @Override
                    public void operationComplete(ChannelFuture future) throws Exception {
                        if (future.isSuccess()) {
                            logger.info("data service. offline message: {}", message.toString());
                            deleteMessages(message.getAppId(), message.getId(), message.getFromId());
                        } else {
                            logger.error("data service. offline message error: {}", future.cause());
                            future.cause().printStackTrace();
                        }
                    }
                });
            });
        }
    }

    public final void sendGroupMessage(List<GroupMessageRequest> groupMessages, Channel channel){
        if(!CollectionUtils.isEmpty(groupMessages)){
            groupMessages.forEach((groupMessage) -> { channel.writeAndFlush(new TextWebSocketFrame(groupMessage.toString())); });
        }
    }

    public abstract List<MessageRequest> getMessages(String appId, String receiverId);

    public abstract void deleteMessages(String appId, String id, String fromId);

    public abstract void deleteGroupMessages(String appId, String id, String fromId);

    public abstract List<GroupMessageRequest> getGroupMessages(String appId, String groupId);
}
